﻿using EventBus.Common;
using System;
using System.Collections.Generic;
using System.Text;
using Microsoft.Extensions.DependencyInjection;
using Confluent.Kafka;
using System.Threading.Tasks;

namespace EventBus.Kafka
{
    public class KafkaSubscriber : IKafkaSubscriber
    {
        private readonly IkafkaConnection _kafkaConnection;
        public KafkaSubscriber(IkafkaConnection kafkaConnection)
        {
            _kafkaConnection = kafkaConnection;
        }

        /// <summary>
        /// 订阅Command
        /// </summary>
        /// <typeparam name="TCommand">命令</typeparam>
        /// <param name="serviceProvider">依赖注入的实例</param>
        /// <param name="topic">主题</param>
        /// <param name="groupId">消费组Id</param>
        public Task SubscriberCommand<TCommand>(IServiceProvider serviceProvider, string topic, string groupId) where TCommand : ICommand
        {
            var commandHandlerImpl = serviceProvider.GetService<ICommandHandler<TCommand>>();
            var consumer = _kafkaConnection.GetConsumer(groupId);
            //订阅指定Topic中的消息
            consumer.Subscribe(topic);

            while (true)
            {
                try
                {
                    var consumeResult = consumer.Consume();
                    var body = consumeResult.Message.Value;
                    var message = Encoding.Default.GetString(body);
                    var command = SwifterJsonSerializer.DeserializeObject<TCommand>(message);
                    commandHandlerImpl.HandleAsync(command).GetAwaiter().GetResult();
                    consumer.Commit(consumeResult);
                }
                catch (ConsumeException e)
                {
                    Console.WriteLine($"Error occured: {e.Error.Reason}");
                }
                catch (Exception e)
                {
                    Console.WriteLine(e.Message);
                }
            }
        }

    }
}
